package org.study.mapreduce;

//package helloworld;

import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WordCount {

	private final static Logger LOG = LoggerFactory.getLogger(WordCount.class);

	static int count = 1;

	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();
		@Override
		public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) {
			String line = value.toString();
			System.err.println("===================== map work " + count++ + "======================");
			List<String> list = Arrays.asList(line.split("\\s|\\W"));
			Iterator<String> it = list.iterator();
			while (it.hasNext()) {
				word.set(it.next());
				try {
					output.collect(word, one);
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			/*
			 * StringTokenizer tokenizer = new StringTokenizer(line); while
			 * (tokenizer.hasMoreTokens()) { word.set(tokenizer.nextToken());
			 * try { output.collect(word, one); } catch (IOException e) {
			 * e.printStackTrace(); } }
			 */
		}
	}

	public static class Reduce extends MapReduceBase implements org.apache.hadoop.mapred.Reducer<Text, IntWritable, Text, IntWritable> {
		@Override
		public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) {
			int sum = 0;
			while (values.hasNext()) {
				int value = values.next().get();
				LOG.info("key-{} = value-{}", key, value);
				sum += value;
			}
			try {
				output.collect(key, new IntWritable(sum));
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}

	public static void main(String[] args) throws IOException {
		init();

		JobConf conf = new JobConf(WordCount.class);
		conf.setJobName("wordcount");

		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(IntWritable.class);

		conf.setMapperClass(Map.class);
		conf.setReducerClass(Reduce.class);

		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		String[] ioArgs = new String[] { "hdfs://10.1.130.142:9300/test/input/LICENSE.txt", "hdfs://10.1.130.142:9300/test/out/result" };
		String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs();

		FileInputFormat.setInputPaths(conf, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(conf, new Path(otherArgs[1]));

		try {
			JobClient.runJob(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	private static void init() {
		forceLoadJNILib();
		System.setProperty("hadoop.home.dir", System.getProperty("user.dir") + "/src/main/resources");
	}

	private static void forceLoadJNILib() {
		try {
			System.load(System.getProperty("user.dir") + "/src/main/resources/bin/hadoop.dll");
		} catch (UnsatisfiedLinkError e) {
			System.err.println("Native code library failed to load.\n" + e);
			System.exit(1);
		}
	}
}
